package com.briup.MR.Sort.TotalSort;

import com.briup.MR.Sort.normal.DescSort;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.net.URI;

/**
 * yarn jar HdfsClientDemo-1.0-SNAPSHOT.jar com.briup.MR.Sort.TotalSort.TotalSortMR -D input=/user/hdfs/wea.txt -D output=/user/hdfs/w_result3
 */
public class TotalSortMR extends Configured implements Tool {
    public int run(String[] args) throws Exception {
        Configuration conf=getConf();
        String input=conf.get("input");
        String output=conf.get("output");
        conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator",",");
        Job job= Job.getInstance(conf);
        job.setJarByClass(this.getClass());
        job.setJobName("TotalSort");

        job.setMapOutputKeyClass(Text.class);

        job.setSortComparatorClass(DescSort.class);

        job.setNumReduceTasks(2);

        job.setInputFormatClass(KeyValueTextInputFormat.class);
        KeyValueTextInputFormat.addInputPath(job,new Path(input));
        TextOutputFormat.setOutputPath(job,new Path(output));
        //设置全局排序
        job.setPartitionerClass(TotalOrderPartitioner.class);
        //第一个参数抽样因子，比例
        //第二个参数抽样的个数
        //注意：第一个和第二个参数谁先完成结束
        //第三个参数划分的区间
        InputSampler.Sampler sample=
                new InputSampler.RandomSampler(0.8,100,
                        4);
        InputSampler.writePartitionFile(job,sample);
        //获取分区的间隔点路径
        String path=TotalOrderPartitioner.getPartitionFile(conf);
        System.out.println(path+"-----------------");
        //分发标准给所有的map
        job.addCacheFile(new URI(path));
        return job.waitForCompletion(true)?0:-1;
    }

    public static void main(String[] args) throws Exception {
        System.exit(
                new ToolRunner().run(
                        new TotalSortMR(),args));
    }
}